package com.atguigu.ct.analysis.io;

import com.atguigu.ct.common.util.JDBCUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Description: MySQL数据格式化输入对象
 * @Author: yyx
 * @Create: 2023/5/5 19:52
 * @Version: 8
 */
public class MySQLRedisTextOutPutFormat extends OutputFormat<Text,Text> {
    //getRecordWriter的静态内部类
    protected static class MySQLRecordWriter extends RecordWriter<Text,Text>{
        private Connection connection = null;
        private Jedis jedis = null;

        //获取资源
        public MySQLRecordWriter() {
            connection = JDBCUtil.getConnection();
            jedis = new Jedis("hadoop102",6379);
        }

        /**
         * 输出数据（Text text, Text text2）分别为key和value
         * @param text
         * @param text2
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void write(Text text, Text text2) throws IOException, InterruptedException {
            String[] values = text2.toString().split("_");
            String sumCall = values[0];
            String sumDuration = values[1];
            String[] keys = text.toString().split("_");
            String tel = keys[0];
            String date = keys[1];
            PreparedStatement pstat = null;
            try {
                String insertSQL = "insert into ct_call(telid,dateid,sumcall,sumduration) values(?,?,?,?)";
                pstat = connection.prepareStatement(insertSQL);
                pstat.setInt(1,Integer.parseInt(jedis.hget("ct_user",tel)));//转为整形
                pstat.setInt(2,Integer.parseInt(jedis.hget("ct_date",date)));
                pstat.setInt(3,Integer.parseInt(sumCall));
                pstat.setInt(4,Integer.parseInt(sumDuration));
                pstat.executeUpdate();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                if(pstat != null){
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        /**
         * 释放资源
         * @param taskAttemptContext
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if(connection != null){
                try {
                    connection.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
            if(jedis != null){
                jedis.close();
            }
        }
    }
    @Override
    public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MySQLRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {
    }
    private FileOutputCommitter committer = null;
    public static Path getOutputPath(JobContext job){
        String name = job.getConfiguration().get(FileOutputFormat.OUTDIR);
        return name == null ? null : new Path(name);
    }
    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        if(committer == null){
            Path output = getOutputPath(taskAttemptContext);
            committer = new FileOutputCommitter(output,taskAttemptContext);
        }
        return committer;
    }
}
